//
// file : ring_buf.c
// date : 4/4/05
//
#include <stdlib.h>
#include <string.h>
#include "ring_buf.h"

/****************************************************************************
**                                                                         **
**  ring buffer                                                            **
**                                                                         **
** ring buffer provides a one-way data passing mechanism between two       **
** threads. it is a linear block of memory whose header contains the       **
** start and the end of the message queue. the writer thread can alter the **
** end of the queue and the reader thread alters the start of the queue.   **
** because the alterations are atomic, which are simply single assignments,** 
** data in the queue maintain their integrity while reader and writer can  **
** operate unobstructively.                                                **
**                                                                         **
** a message is prefixed with a message header that contains message       **
** length in bytes. a reader parsed all message in the ring by reading     **
** message headers sequencially.                                           **
**                                                                         **
** a read operation causes ring to move, unless the provided buffer is too **
** small to hold the current message, in which case the needed buufer size **
** is returned. the caller must always compare the returned value with the **
** message buffer length to make sure the buffer is filled.                **
**                                                                         **
** since read buffer is filled before the ring moves its queue start,      **
** writer can not overwrite a message been read.                           **
**                                                                         **
** similarly, writer updates queue end after the whole message is copied   **
** into the ring.                                                          **
****************************************************************************/


#define Z_ALIGN(a, m) ((a)%(m)?((m)-(a)%(m)):0)

/****************************************************************************
**                                                                         **
** long next_anchor(msgring_t *rb, long pos_now, long msglen)              **
**                                                                         **
**   calculate the next rpos/wpos position with proper alignment           ** 
**   after read/write operations                                           **
** pos_now is the current position.                                        **
** msglen is the message length to read/write                              **
** return value is the next position after read/write                      **
****************************************************************************/
long next_anchor(ring_t *rb, long pos_now, long msglen)
{
    long headlen = sizeof(msg_t);

    long pos_next;

    /*
    ** check if we have enough room
    */
    if( rb->end < headlen + msglen )
    {
        return (-1);
    }

    /*
    ** note that pos_now is headlen aligned
    */
    if( rb->end - pos_now >= headlen )
    {
        pos_next = pos_now + headlen + msglen;
    }
    else /* wrap */
    {
        pos_next = headlen + msglen;
    }

    pos_next += Z_ALIGN(pos_next, headlen);

    if( pos_next > rb->end )
    {
        pos_next -= rb->end;

        pos_next += Z_ALIGN(pos_next, headlen);
    }
    else if( pos_next + headlen > rb->end )
    {
        pos_next = 0;
    }

    return pos_next;
}

/****************************************************************************
**                                                                         **
** read from msgring                                                       **
**                                                                         **
** return value is message length, including message header.               **
** if 'buf->msglen' is smaller than the message length,                    **
** 'buf' is not filled and ring is not moving                              **
****************************************************************************/
long  rb_read(ring_t *rb, msg_t *buf)
{
    long    tail;

    msg_t  *msg;

    u8_t   *b, *s;

    long    headlen = sizeof(msg_t);

    if(rb->rpos == rb->wpos)
    {
        return 0;  /* nothing to read */
    }

    /*
    ** get read head
    */
    msg = (msg_t*)(rb->start + rb->rpos);

    /*
    ** make sure caller provides enough buffer
    */
    if( 0 == buf || msg->len > buf->len )
    {
        return msg->len + headlen;
    }

    /*
    ** copy message 
    */
    b = (u8_t*)buf + headlen;

    s = (u8_t*)msg + headlen;

    *buf = *msg;

    /*
    ** message data may wrap at the end of linear buffer
    */
    tail = rb->end - rb->rpos - headlen;

    //assert(tail >= 0);

    if( tail > 0 )
    {
        if( tail < msg->len)
        {
            memcpy(b, s, tail);

            b += tail;
        }
        else
        {
            if( 0 < msg->len ) memcpy(b, s, msg->len);
        }
    }

    if( tail < msg->len )
    {
        memcpy(b, rb->start, msg->len - tail);
    }

    /*
    ** move read pointer, adjust to multiple of sizeof(msg_t)
    */
    rb->rpos = next_anchor(rb, rb->rpos, msg->len);

    return msg->len + headlen;
}


/****************************************************************************
**                                                                         **
** write msg to msgring                                                    **
**                                                                         **
** return value is message length that is written to ring buffer           **
** if no enough room available, return 0                                   **
****************************************************************************/
long  rb_write(ring_t *rb, msg_t *msg)
{
    long   wpos, tail;

    msg_t *buf;

    u8_t  *b, *s;

    long  headlen = sizeof(msg_t);

    wpos = next_anchor(rb, rb->wpos, msg->len);

    /*
    ** reject messages that is too long to fit message ring
    */
    if( wpos < 0 )
    {
        return (-1);
    }

    /*
    ** check if we have room for it. note that if 
    ** rb->wpos == rb->rpos then the ring is considered
    ** empty. otherwise if 
    */
    if( rb->wpos > rb->rpos )
    {
        /*
        ** wrap and pass over rb->rpnt
        */
        if( wpos < rb->wpos && wpos >= rb->rpos )
        {
            return 0;
        }
    }
    else if( rb->wpos < rb->rpos )
    {
        /*
        ** pass over rb->rpnt or wrap
        */
        if( wpos >= rb->rpos || rb->wpos > wpos )
        {
            return 0;
        }
    }

    /*
    ** copy message data
    */
    buf = (void *)(rb->start + rb->wpos);

    *buf = *msg;

    b = (u8_t *)buf + headlen;

    s = (u8_t *)msg + headlen;

    tail = rb->end - rb->wpos - headlen;

    //assert(tail >= 0);

    if(tail > 0)
    {
        if( tail < msg->len)
        {
            memcpy(b, s, tail);

            s += tail;
        }
        else
        {
            if( 0 < msg->len ) memcpy(b, s, msg->len);
        }
    }

    if(tail < msg->len)
    {
        memcpy(rb->start, s, msg->len - tail);
    }

    rb->wpos = wpos;

    return msg->len + headlen;
}


/****************************************************************************
**                                                                         **
** abandon a message at the the ring buffer head                           **
**                                                                         **
** return value is message length that is skipped                          **
**                                                                         **
****************************************************************************/

long rb_yank(ring_t *rb)
{
    msg_t  *msg;

    if(rb->rpos == rb->wpos)
    {
        return 0;  /* nothing to remove */
    }

    /*
    ** get read head
    */
    msg = (msg_t*)(rb->start + rb->rpos);

    /*
    ** move read pointer, adjust to multiple of sizeof(msg_t)
    */
    rb->rpos = next_anchor(rb, rb->rpos, msg->len);

    return msg->len + sizeof(msg_t);
}

///////////////////////////////////////////////////////////////////////////

